#include "server.h"
#include "iothreadpool.h"
#include <signal.h>

#define CLIENT_MAX_NUM 1024


Server server;

void Server::initServer(int port)
{
    signal(SIGPIPE, SIG_IGN);

    std::string logpath = "./log/server.log";
    logger = tcalloc(qLogger, logpath);
    port_ = port;

    sockfd_ = connection::Socket();
    if(sockfd_ < 0)
      exit(1);
  
    if(connection::SetSockOpt(sockfd_) < 0)
      exit(1);

    if(connection::Bind(sockfd_, port_) < 0)
      exit(-1);

    if(connection::Listen(sockfd_) < 0)
      exit(-1);

    el_ = tcalloc(eventLoop, CLIENT_MAX_NUM);
    el_->createFileEvent(sockfd_, AE_READABLE, eventLoop::acceptEvent, nullptr);

    io_thread_ = IOThreadPool::getInstance();
    io_thread_->InitThreadIO();
}

void Server::clientInstallWriteHandler(connection* conn)
{
    if(!(conn->flags_ & CONN_PENDING_WRITE))
    {
        conn->flags_ |= CONN_PENDING_WRITE;
        clients_pending_write_.push_back(conn);
    }
}

void Server::freeClientAsync(connection *c)
{
    if(c->flags_ & CONN_CLOSE_ASAP) return;

    c->flags_ |= CONN_CLOSE_ASAP;
    if(io_thread_num_ == 1)
    {
      clients_to_close_.push_back(c);
      return;
    }
    
    //std::cout << "freeClientAsync0: " << clients_to_close_.size() << std::endl;
    static pthread_mutex_t async_free_queue_mt = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_lock(&async_free_queue_mt);
    //std::cout << "freeClientAsync1: " << clients_to_close_.size() << std::endl;
    clients_to_close_.push_back(c);
    pthread_mutex_unlock(&async_free_queue_mt);
    //std::cout << "freeClientAsync2: " << clients_to_close_.size() << std::endl;
}

void Server::freeClient(connection* conn)
{
    std::cout << "close fd: " << conn->fd_ << "conn: " << conn << std::endl;
    //删除epoll事件
    server.el_->deleteFileEvent(conn->fd_, AE_READABLE | AE_WRITABLE);

    // std::list<replyBlock*>::iterator it = conn->reply_.begin();
    // for(; it != conn->reply_.end(); ++it)
    // {
    //     if(*it)
    //     {
    //         std::cout << *it << std::endl;
    //         free(*it);
    //         *it = nullptr;
    //     }
    // }
    // conn->reply_.clear();

    //delete conn;
    tcfree(connection, conn);
    conn = nullptr;
}


int Server::freeClientsInAsyncFreeQueue()
{
    int freed = 0;
    std::list<connection*>::iterator it = clients_to_close_.begin();
    for(; it != clients_to_close_.end(); )
    {
        connection* conn = *it;
        if(!(conn->flags_ & CONN_CLOSE_ASAP))
        {
            it++;
            continue;
        }
        
        conn->flags_ &= ~CONN_CLOSE_ASAP;
        freeClient(conn);
        it = clients_to_close_.erase(it);
        freed++;
    }
    clients_to_close_.clear();
    return freed;
}


int Server::handleClientsWithPendingReadsUsingThreads()
{
    if(!io_thread_active_) return 0;
    int processed = clients_pending_read_.size();
    if(processed == 0) return 0;

    //std::cout << "Reads 分配任务1: " << processed << std::endl;

            //将任务从全局队列负载均衡到io线程队列中
    int item_id = 0;
    std::list<connection*>::iterator it = clients_pending_read_.begin();
    for(; it != clients_pending_read_.end(); ++it)
    {
        (*it)->flags_ &= ~CONN_PENDING_READ;

        int target_id = item_id % io_thread_num_;
        io_thread_->pushThreadList(target_id, *it);
        item_id++;
    }

    //std::cout << "Reads 分配任务1: " << 0 << ":" << IOThreadPool::io_thread_list_[0].size() << std::endl;
    io_thread_->setThreadOperation(IO_THREADS_OP_READ);
    for(int i = 1; i < io_thread_num_; ++i)
    {
        int count = io_thread_->getThreadListSize(i);
        //std::cout << "Reads 分配任务1: " << i << ":" << count << std::endl;
        io_thread_->setIOPendingCount(i, count);
    }
    
    //主线程处理任务
    io_thread_->handleReadTask(0);

    //等待io线程处理完
    while(1)
    {
        int pendings = 0;
        for(int i = 1; i < io_thread_num_; ++i)
        {
            pendings += io_thread_->getIOPendingCount(i);
        }
        if(pendings == 0)
            break;
    }
    clients_pending_read_.clear();

    return processed;

}

int Server::handleClientsWithPendingWrites()
{
    //std::cout << "handleClientsWithPendingWrites size: " << clients_pending_write_.size() << std::endl;
    std::list<connection*>::iterator it = clients_pending_write_.begin();
    for(; it != clients_pending_write_.end(); )
    {
        connection* conn = *it;

        conn->flags_ &= ~CONN_PENDING_WRITE;
        //std::cout << "handleClientsWithPendingWrites: " << *it << std::endl;
        io_thread_->writeToClient(conn);
        clients_pending_write_.erase(it++);

        //TODO:防止write时出现EAGAIN直接返回，而不是循环继续发送，导致数据没有发送
        //这里添加连接为写事件，在epoll进行检测
        if(conn->hasPendingReplies())
        {
            if(conn->connnSocketSetWriteHandler(IOThreadPool::sendReplyToClient) < 0)
                server.freeClientAsync(conn);
        }
        else 
        {
            //http加 短连接
            server.freeClientAsync(conn);
        }
    }

}

bool Server::stopThreadIOIfNeeded()
{
    int pending = clients_pending_write_.size();

    if(io_thread_num_ == 1) return true;

    if(pending < 2)
    {
        if(io_thread_active_)
            io_thread_->stopThreadIO();
        return true;
    }
    else
    {
        return false;
    }
}

int Server::handleClientsWithPendingWritesUsingThreads()
{
    int processed = clients_pending_write_.size();
    if(processed == 0) 
    {
        //std::cout << "handleClientsWithPendingWritesUsingThreads size: " << processed << std::endl;
        return 0;
    }
    if(io_thread_num_ == 1 || stopThreadIOIfNeeded())
    {
        //因为在处理完读事件后。会直接将写时间加到全局队列中，所以这里需要单独处理
        return handleClientsWithPendingWrites();
    }

    if(!io_thread_active_)
        io_thread_->startThreadIO();
    
    //std::cout << "Writes 分配任务2: " << processed << std::endl;
    //将任务从全局队列负载均衡到io线程队列中
    io_thread_->setThreadOperation(IO_THREADS_OP_WRITE);
    int item_id = 0;
    std::list<connection*>::iterator it = clients_pending_write_.begin();
    for(; it != clients_pending_write_.end(); ++it)
    {
        (*it)->flags_ &= ~CONN_PENDING_WRITE;

        int target_id = item_id % io_thread_num_;
        io_thread_->pushThreadList(target_id, *it);
        item_id++;
    }
    //std::cout << "Writes 分配任务2: " << 0 << ":" << IOThreadPool::io_thread_list_[0].size() << std::endl;
    for(int i = 1; i < io_thread_num_; ++i)
    {
        int count = io_thread_->getThreadListSize(i);
        //std::cout << "Writes 分配任务2: " << i << ":" << count << std::endl;
        io_thread_->setIOPendingCount(i, count);
    }
    
    //主线程处理任务
    io_thread_->handleWriteTask(0);

    //等待io线程处理完
    while(1)
    {
        int pendings = 0;
        for(int i = 1; i < io_thread_num_; ++i)
        {
            pendings += io_thread_->getIOPendingCount(i);
        }
        if(pendings == 0)
            break;
    }

     /* Run the list of clients again to install the write handler where
     * needed. */
    it = clients_pending_write_.begin();
    for(; it != clients_pending_write_.end(); ++it)
    {
        connection* conn = *it;

        if(conn->hasPendingReplies())
        {
            //设置写回调函数
            if(conn->connnSocketSetWriteHandler(IOThreadPool::sendReplyToClient) < 0)
            {
                freeClientAsync(conn);
            }
        }
    }

    clients_pending_write_.clear();

    return processed;
}


int main()
{
  server.initServer(9090);

  server.stop_ = 0;
  while(!server.stop_)
  {
    server.el_->processEvents();
  }
}